package com.sdp.irrigation.server.kafka.consumer;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import com.alibaba.fastjson.JSON;
import com.sdp.irrigation.protocol.CommPackage;
import com.sdp.irrigation.server.netty.response.IResponseHandle;
import com.sdp.irrigation.utils.CommConstants;
import com.sdp.irrigation.utils.JsonUtils;

import cn.hutool.core.codec.Base64;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class KafkaBatchConsumer {

	private IResponseHandle jsonResponseHandle;

	public KafkaBatchConsumer(IResponseHandle jsonResponseHandle) {
		this.jsonResponseHandle = jsonResponseHandle;
	}
	
	private void commonBiz(List<String> records) {
		if (records.size() > 0) {
			for (String msg : records) {
				CommPackage commPackage = JsonUtils.parseBean(Base64.decodeStr(msg), CommPackage.class);
				jsonResponseHandle.sendMessage(commPackage.getDevSn(), commPackage);
			}
			
		}
	}
	
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "report")
	public void commandEqStatusQuery(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//开关阀门
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "200")
	public void commandValve(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//打开水泵
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "180")
	public void commandEqOpen(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//关闭水泵
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "190")
	public void commandEqClose(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//加水
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "920")
	public void commandWaterSupply(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//搅拌
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "930")
	public void commandAgitation(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//施肥
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "950")
	public void commandFertigation(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//排沙
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "910")
	public void commandDesilting(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//结账
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "210")
	public void commandSettle(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//余额查询结果
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "220")
	public void commandQueryBalance(List<String> records, Acknowledgment ack) {
		try {
			commonBiz(records);
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
	//用户帧应答
	@KafkaListener(topics = "${sdp.server.down-topic-prefix}" + "50")
	public void commandUserResponse(List<String> records, Acknowledgment ack) {
		try {
			if (records.size() > 0) {
				for (String msg : records) {
					@SuppressWarnings("unchecked")
					Map<String, Object> map = JSON.parseObject(Base64.decodeStr(msg), LinkedHashMap.class);
					jsonResponseHandle.sendMessage(map.get("devSn").toString(), map);
				}
				
			}
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			ack.acknowledge();
		}
	}
	
}
